#ifndef __VZMQ_H__
#define __VZMQ_H__

#include "zmq.hpp"
#include <google/protobuf/message.h>
#include <iostream>
#include <functional>
#include <string>
#include <thread>
#include <memory>
#include <atomic>

namespace vzmq
{
	/// @brief check topic is available or not
	/// @param topic current topic, if unavailable return old topic
	/// @param address current topic, if unavailable return ""
	/// @return bool
	static bool topicIsAvailable(std::string &topic, std::string &address)
	{
		if (topic.rfind("tcp") != 0 && topic.rfind("udp") != 0)
		{
			std::string c = topic;
			std::replace(c.begin(), c.end(), '/', '_');
			address = "ipc:///var/run/" + c;
			return true;
		}
		else
		{
			address = topic;
		}
		return true;
	}

	template <typename MessageT>
	class VzmqSubscriber
	{
	public:
		/// @brief Constructor function
		/// @param topic subscribe topic
		/// @param context zmq::context_t
		/// @param cb callback function,if get message by topic, callback function will be called
		VzmqSubscriber(const std::string &topic, zmq::context_t *context, std::function<void(const std::shared_ptr<MessageT> &)> cb)
			: _topic(topic), _context(context)
		{
			std::cout << "VzmqSubscriber created --" << std::endl;
			if (!topicIsAvailable(_topic, _serverAddress))
				return;
			std::cout << _topic << "  " << _serverAddress << std::endl;
			_loopExit = false;
			_thread = std::make_unique<std::thread>(&VzmqSubscriber::loop, this, cb);
			std::cout << "VzmqSubscriber created ++" << std::endl;
		}

		~VzmqSubscriber() { disconnect(); }

	private:
		void loop(std::function<void(const std::shared_ptr<MessageT> &)> cb)
		{
			zmq::socket_t socket{};
			zmq::socket_t signalStop{};
			connectServer(socket, signalStop);
			std::vector<zmq::pollitem_t> p = {{signalStop, 0, ZMQ_POLLIN, 0}, {socket, 0, ZMQ_POLLIN, 0}};
			while (!_loopExit.load(std::memory_order_acquire))
			{
				zmq::poll(p.data(), p.size(), -1);
				if (p[0].revents & ZMQ_POLLIN)
				{
					continue;
				}
				else if (p[1].revents & ZMQ_POLLIN)
				{
					zmq::message_t dataMessage;
					socket.recv(&dataMessage);
					if (!_loopExit && cb)
					{
						std::string serializedData(static_cast<char *>(dataMessage.data()), dataMessage.size());
						auto recMsg = std::make_shared<MessageT>();
						recMsg->ParseFromString(serializedData);
						cb(recMsg);
					}
				}
			}
		}

		void connectServer(zmq::socket_t &socket, zmq::socket_t &signalStop)
		{
			socket = zmq::socket_t(*_context, ZMQ_SUB);
			std::cout << _serverAddress << " ++ " << std::endl;
			socket.connect(_serverAddress);
			std::cout << _serverAddress << std::endl;
			socket.set(zmq::sockopt::subscribe, "");
			/// connect signal stop
			char signalStopAddr[255];
			std::hash<std::string> hasher;
			auto hash = hasher(_serverAddress);
			snprintf(signalStopAddr, 255, "inproc://%p_%d_%zu", (void *)this, rand(), hash);
			signalStop = zmq::socket_t(*_context, ZMQ_PAIR);
			signalStop.bind(signalStopAddr);
			_signalStopAddress = std::string(signalStopAddr);
			std::cout << _signalStopAddress << std::endl;
		}

		void interruptReceive()
		{
			zmq::message_t msg{1};
			memcpy(msg.data(), "0", 1);
			zmq::socket_t doSignal(*_context, ZMQ_PAIR);
			doSignal.connect(_signalStopAddress);
			doSignal.send(msg, zmq::send_flags::dontwait);
		}
		void disconnect()
		{
			std::cout << "VzmqSubscriber disconnect ++" << std::endl;
			_loopExit = true;
			interruptReceive();
			if (_thread)
			{
				_thread->join();
				_thread = nullptr;
			}
			std::cout << "VzmqSubscriber disconnect --" << std::endl;
		}

	private:
		std::string _topic;
		std::string _serverAddress;
		std::string _signalStopAddress;
		zmq::context_t *_context;
		std::unique_ptr<std::thread> _thread{nullptr};
		std::atomic<bool> _loopExit{true};
	};

	template <typename MessageT>
	class VzmqPublisher
	{
	public:
		VzmqPublisher(const std::string &topic, zmq::context_t &context)
			: _topic(topic), _publisher(context, ZMQ_PUB)
		{
			std::cout << "VzmqPublisher created --" << std::endl;
			if (!topicIsAvailable(_topic, _address))
				return;
			_publisher.bind(_address);
			std::cout << _topic << "  " << _address << std::endl;
			std::cout << " VzmqPublisher bind address: " << _address << std::endl;
			_running = true;
			std::cout << "VzmqPublisher created ++" << std::endl;
		}

		void publish(const std::shared_ptr<MessageT> &message)
		{
			if (!_running.load(std::memory_order_acquire))
				return;
			std::string serializedMessage;
			message->SerializeToString(&serializedMessage);
			zmq::message_t dataMessage(serializedMessage.size());
			memcpy(dataMessage.data(), serializedMessage.data(), serializedMessage.size());
			_publisher.send(dataMessage,zmq::send_flags::none);
			
		}

		~VzmqPublisher() { destroy(); }

	private:
		void destroy()
		{
			std::cout << "VzmqPublisher destroyed ++" << std::endl;
			_publisher.close();
			std::cout << "VzmqPublisher destroyed --" << std::endl;
		}

	private:
		std::string _topic;
		std::string _address;
		zmq::socket_t _publisher;
		std::atomic<bool> _running{false};
	};

	class VzmqContext
	{
	public:
		VzmqContext()
		{
			_context.set(zmq::ctxopt::ipv6, 1); // enabele IPV6
		};
		template <typename MessageT>
		std::shared_ptr<VzmqPublisher<MessageT>> createPublisher(const std::string &topic)
		{
			return std::make_shared<VzmqPublisher<MessageT>>(topic, _context);
		}
		template <typename MessageT>
		std::shared_ptr<VzmqSubscriber<MessageT>> createSubscribler(const std::string &topic, const std::function<void(const std::shared_ptr<MessageT> &)> &cb)
		{
			return std::make_shared<VzmqSubscriber<MessageT>>(topic, &_context, cb);
		}

	private:
		const std::string _address;
		zmq::context_t _context{};
	};

}

#endif // __VZMQ_H__